package com.exd.flink.sink.business;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import com.exd.flink.domain.VDataBean;
public class VdataShardSink extends RichSinkFunction<VDataBean>{
	private static final long serialVersionUID = 1L;
    private static org.apache.hadoop.conf.Configuration configuration;
    private static Connection connection= null;
    private static BufferedMutator bufferedMutator;
    private String tablename;
    
    public VdataShardSink(String tablename) {
    	this.tablename=tablename;
    }
 
    @Override
    public void open(Configuration parameters) throws Exception {
        configuration=HBaseConfiguration.create();
		configuration.set("hbase.zookeeper.quorum", "192.168.124.76");
		configuration.set("hbase.zookeeper.property.clientPort", "2181");
        try {
             connection = ConnectionFactory.createConnection(configuration);
             BufferedMutatorParams params = new BufferedMutatorParams(TableName.valueOf(tablename));
             //缓存大小
             params.writeBufferSize(2*1024*1024);
             //最大时间
             params.setWriteBufferPeriodicFlushTimeoutMs(5 * 1000L);
             bufferedMutator = connection.getBufferedMutator(params);
        }catch (Exception e){
            e.printStackTrace();
        }
 
    }
 
    @Override
    public void close() throws Exception {
        if (bufferedMutator != null) {
        	bufferedMutator.close();
        }
        if (connection != null) {
            connection.close();
        }
 
    }
 
    @Override
    public void invoke(VDataBean dataBean, Context context) throws Exception {
    	String rowKey=dataBean.getDeployId()+dataBean.getServiceId()+dataBean.getVin()+dataBean.getCollectTime();
        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("serviceId"),Bytes.toBytes(dataBean.getServiceId()));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("vin"),Bytes.toBytes(dataBean.getVin()));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("collectTime"),Bytes.toBytes(dataBean.getCollectTime()));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("packetSequence"),Bytes.toBytes(dataBean.getPacketSequence()));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("sliceNum"),Bytes.toBytes(dataBean.getSliceNum()));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("sliceLast"),Bytes.toBytes(dataBean.getSliceLast()));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("key"),Bytes.toBytes(dataBean.getKey()));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("samplingTime"),Bytes.toBytes(dataBean.getSamplingTime()));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("tboxSendTimestamp"),Bytes.toBytes(dataBean.getTboxSendTimestamp()));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("receiveTimestamp"),Bytes.toBytes(dataBean.getReceiveTimestamp()));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("methodId"),Bytes.toBytes(dataBean.getMethodId()));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("majorVersion"),Bytes.toBytes(dataBean.getMajorVersion()));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("year"),Bytes.toBytes(dataBean.getYear()));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("month"),Bytes.toBytes(dataBean.getMonth()));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("day"),Bytes.toBytes(dataBean.getDay()));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("hour"),Bytes.toBytes(dataBean.getHour()));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("min"),Bytes.toBytes(dataBean.getMin()));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("sec"),Bytes.toBytes(dataBean.getSec()));
        put.addColumn(Bytes.toBytes("vdata"),Bytes.toBytes("data"),Bytes.toBytes(dataBean.getData()));
        bufferedMutator.mutate(put);
        // 指定时间内的数据强制刷写到hbase
        bufferedMutator.flush();
 
 
    }
}